// Copyright 2009 StarCite Inc.  All rights reserved.
// MongoDBClient.java 4:57:08 PM danny.liu
package com.billowsoft.blog.dao;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.bson.types.ObjectId;

import com.mongodb.AggregationOutput;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MapReduceCommand;
import com.mongodb.MapReduceOutput;
import com.mongodb.Mongo;
import com.mongodb.MongoException;

/**
 * @author danny.liu Jun 18, 2012
 * @version 1.0
 * 
 */
public class MongoDBClient {

	private static Log LOG = LogFactory.getLog(MongoDBClient.class);
	
	private static DB mongoDB;

	public MongoDBClient(String host, int port, String dbName) {
		try {
			Mongo mongo = new Mongo(host, port);
			mongoDB = mongo.getDB(dbName);
		}
		catch (UnknownHostException e) {
			throw new RuntimeException("Can not connect to mongoDB.", e);
		}
		catch (MongoException e) {
			throw new RuntimeException("Can not connect to mongoDB.", e);
		}
	}

	public <T> String saveObject(String collectionName, T object) {
		DBCollection coll = mongoDB.getCollection(collectionName);
		ObjectId id = new ObjectId();
		BasicDBObject dbObject = MongoDBUtil.convertObjectToMongoObject(object);
		dbObject.append("_id", id);
		coll.insert(dbObject);
		return id.toString();
	}

	public <T> void saveObjects(String collectionName, List<T> objects) {
		DBCollection coll = mongoDB.getCollection(collectionName);
		List<DBObject> mongoObjects = new ArrayList<DBObject>(objects.size());
		for (Object obj : objects) {
			mongoObjects.add(MongoDBUtil.convertObjectToMongoObject(obj));
		}
		coll.insert(mongoObjects);
	}

	public Map<String, Object> queryDoc(String collectionName, DBObject criteria) {

		DBCollection coll = mongoDB.getCollection(collectionName);
		DBObject result = coll.findOne(criteria);
		return result.toMap();
	}

	public List<Map<String, Object>> queryDocs(String collectionName, Map<String, String> criteries, int limit, int start) {

		List<Map<String, Object>> result = new ArrayList<Map<String, Object>>();
		DBCollection coll = mongoDB.getCollection(collectionName);

		DBCursor cursor = coll.find(MongoDBUtil.convertObjectToMongoObject(criteries)).skip(start).limit(limit);
		while (cursor.hasNext()) {
			DBObject obj = cursor.next();
			result.add(obj.toMap());
		}

		return result;
	}

	public long queryCount(String collectionName, Map<String, Object> criteries) {

		DBCollection coll = mongoDB.getCollection(collectionName);
		return coll.getCount(MongoDBUtil.convertObjectToMongoObject(criteries));
	}

	public <T> void update(String collectionName, Map<String, String> criteria, T newObject) {

		DBCollection coll = mongoDB.getCollection(collectionName);
		coll.update(new BasicDBObject(criteria), MongoDBUtil.convertObjectToMongoObject(newObject));
	}

	public void update(String collectionName, Map<String, String> criteria, Map<String, String> newValues) {

		update(collectionName, criteria, newValues);
	}

	public void delete(String collectionName, Map<String, String> criteria) {

		DBCollection coll = mongoDB.getCollection(collectionName);
		coll.remove(new BasicDBObject(criteria));
	}

	public List<Map> queryWithMapReduce(String collectionName, String mapFunc, String reduceFunc, Map<String, String> criteries, int limit, int start) {
		DBCollection coll = mongoDB.getCollection(collectionName);

		MapReduceCommand command = new MapReduceCommand(coll, mapFunc, reduceFunc, null, MapReduceCommand.OutputType.INLINE,
				MongoDBUtil.convertObjectToMongoObject(criteries));
		command.setLimit(limit);

		MapReduceOutput output = coll.mapReduce(command);
		List<Map> result = new ArrayList<Map>();
		for (DBObject obj : output.results()) {
			result.add(obj.toMap());
		}

		return result;
	}

	public Map queryWithMapReduce(String collectionName, String mapFunc, String reduceFunc, Map<String, String> criteries) {

		return queryWithMapReduce(collectionName, mapFunc, reduceFunc, MongoDBUtil.convertObjectToMongoObject(criteries));
	}

	public Map queryWithMapReduce(String collectionName, String mapFunc, String reduceFunc, DBObject criteriaObj) {

		DBCollection coll = mongoDB.getCollection(collectionName);
		MapReduceCommand command = new MapReduceCommand(coll, mapFunc, reduceFunc, null, MapReduceCommand.OutputType.INLINE, criteriaObj);
		MapReduceOutput output = coll.mapReduce(command);

		List<Map> result = new ArrayList<Map>();
		for (DBObject obj : output.results()) {
			return obj.toMap();
		}

		return null;
	}

	public Map queryAggregation(String collectionName, DBObject queryCriteria, DBObject aggrFields, DBObject aggrFunc) {

		DBCollection coll = mongoDB.getCollection(collectionName);
		// create our pipeline operations, first with the $match
		DBObject match = new BasicDBObject("$match", new BasicDBObject("type", "airfare"));

		// build the $projection operation
		DBObject fields = new BasicDBObject("department", 1);
		fields.put("amount", 1);
		fields.put("_id", 0);
		DBObject project = new BasicDBObject("$project", fields);

		// Now the $group operation
		DBObject groupFields = new BasicDBObject("_id", "$department");
		groupFields.put("average", new BasicDBObject("$avg", "$amount"));
		DBObject group = new BasicDBObject("$group", groupFields);

		// run aggregation
		AggregationOutput output = coll.aggregate(queryCriteria, aggrFields, aggrFunc);
		for(DBObject obj : output.results()){
			return obj.toMap();
		}
		
		return null;
	}
}
